package cassandrawindow;
import java.util.*;
import java.text.*;
import java.text.SimpleDateFormat;
/**
 * @author quyennb
 */
public class HandleCountClick {

     public static void handleData() throws ParseException{
        //get all key from redis]
        List<String> values = new ArrayList<String>();
        List<String> listDel = new ArrayList<String>();
        Map<Integer, String[]> countList = Collections.synchronizedMap(new HashMap<Integer, String[]>());

        String host = "192.168.23.85";
        int port    = 6379;
        String[]    keysHashs ;
        String[]    fields;

        long startTick = System.currentTimeMillis();
        //InitCassandra initCa = new InitCassandra();

        CounterClick cc = new CounterClick(host, port);
        int dbsize = cc.rClient.dbsize();
        Debug.println("amount of keys :" +dbsize);

        keysHashs = cc.rClient.keys("*");

        //lay tung key voi tat ca gia tri ben trong no
        for(int i=0 ;i<keysHashs.length ;i++)
        {
            fields = cc.rClient.hgetall(keysHashs[i]);
            countList.put(i, fields);
            values.clear();
        }

        Set set = countList.entrySet();
        Iterator iter = set.iterator();

        while (iter.hasNext()) {
            Map.Entry entry = (Map.Entry) iter.next();
            int keyMap =(Integer) entry.getKey();
            String[] keyHandle = (String[]) entry.getValue();
            List<String> keyTemp = new ArrayList<String>();
            List<String> valTemp = new ArrayList<String>();

            long checkToDel = 0;

            for (int i = 0 ; i< keyHandle.length ; i++)
            {
                if (i % 2 == 0)
                {
                    keyTemp.add(keyHandle[i]);
                    long valDate = dataToLong(keyHandle[i]);

                    if(valDate >= checkToDel)
                    {
                        checkToDel = valDate;
                    }

                    listDel.add(keyHandle[i]);

                } else {
                    valTemp.add(keyHandle[i]);
                }
            }
            //insert cassandra
            TestThread tt = new TestThread(valTemp, keyTemp, 0, keyTemp.size());
            int amountKeyDel = 0;
            //tao transaction de del tat ca cac key it nhat
             try {
                cc.rClient.multi();
                for(int i = 0 ; i < listDel.size(); i++)
                {
                    long valDate = dataToLong(listDel.get(i));
                    if(checkToDel > valDate)
                    {
                        amountKeyDel++;
                        cc.rClient.hdel(keysHashs[keyMap],listDel.get(i));
                        //Debug.println(keysHashs[keyMap] +" :" + '"' +listDel.get(i)+'"');
                    }
                }
                cc.rClient.exec();
             } catch (Exception ex){
                    Debug.println(ex.getMessage());
            }
            //del list de bo nho cho cai khac
            Debug.println("so luong key da del voi hash "+keysHashs[keyMap] + " la :" + amountKeyDel);
            Debug.println("so luong key con lai voi hash "+keysHashs[keyMap] + "chua del la :" + (listDel.size() - amountKeyDel));
            listDel.clear();
        }
        long finish = System.currentTimeMillis();
        Debug.println("time transaction :"+String.valueOf(finish-startTick));
        Debug.println("sucess");
    }


    public static long longToDate(String str) throws ParseException
    {
        SimpleDateFormat df = (SimpleDateFormat) SimpleDateFormat.getInstance();
        df.applyPattern("dd/MM/yyyy");
       // Date date = df.parse("11/2/2008 00:00:00");

        Date date = df.parse(str);
        //System.out.println(date);
        Calendar cal = new GregorianCalendar();
        cal.setTime(date);
        //System.out.println(cal.getTimeInMillis()/1000);
        return cal.getTimeInMillis()/1000;
    }

    public static long dataToLong(String keyHandle) throws ParseException
    {
        long epoch = 0;
        String[] arr ;
        String[] hoursplit ;
        try {
                arr = keyHandle.split(" ");

                if(arr.length >= 2)
                {
                    hoursplit = arr[1].split("-");
                    String time =  arr[0] + " " + hoursplit[0] + ":00";
                    epoch = new java.text.SimpleDateFormat ("yyyy-MM-dd HH:mm:ss").parse(time).getTime()/1000;
                } else {
                    hoursplit = arr[0].split("-");
                    String time =  hoursplit[0] + "-" + hoursplit[1] + "-" + hoursplit[2];
                    epoch = new java.text.SimpleDateFormat ("yyyy-MM-dd").parse(time).getTime()/1000;
                }
        } catch(Exception ex ) {
            Debug.println(ex.getMessage());
        }

      return epoch;
    }



}
